package com.pw.study.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class $06SqlFile {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        readFile(tEnv);
        writeFile(tEnv);


    }

    private static void writeFile(StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table tb1( id string, ts bigint,vc int) with ( 'connector'='filesystem', " +
                " 'path'='data/file/b','format'='csv')");
        tEnv.sqlQuery("select * from sensor").executeInsert("tb1");
        tEnv.sqlQuery("select * from tb1 where id='sensor_1'").execute().print();
    }

    private static void readFile(StreamTableEnvironment tEnv) {
        tEnv.executeSql("create table sensor(" +
                " id string," +
                " ts bigint," +
                " vc int) with (" +
                " 'connector' = 'filesystem', " +
                " 'path' = 'data/file/sensor.txt'," +
                " 'format' = 'csv' ) ");

        tEnv.sqlQuery("select * from sensor where id='sensor_1'").execute().print();
    }


}
